package com.example.capsule_kafka.listener;


import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.capsule_kafka.bean.ChatRecordBean;
import com.example.capsule_kafka.bean.GroupMessageBean;
import com.example.capsule_kafka.bean.MacAddressToFieldBean;
import com.example.capsule_kafka.bean.RoomBean;
import com.example.capsule_kafka.impl.ChatRecordServiceImpl;
import com.example.capsule_kafka.mapper.GroupMessageMapper;
import com.example.capsule_kafka.mapper.MacAddressToFieldMapper;
import com.example.capsule_kafka.mapper.RoomMapper;
import com.google.gson.Gson;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * @Author: zhuzhiming
 * @Date: 2022/5/16 12:59
 */
@Component
public class ChatRecordListener implements ApplicationListener<ContextRefreshedEvent> {

    @Autowired
    ChatRecordServiceImpl chatRecordService;

    @Resource
    RoomMapper roomMapper;

    @Resource
    MacAddressToFieldMapper macAddressToFieldMapper;

    @Resource
    GroupMessageMapper groupMessageMapper;

    @Autowired
    RestHighLevelClient esClient;


    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {

        List<MacAddressToFieldBean> fieldBeanList = macAddressToFieldMapper.selectList(new QueryWrapper<MacAddressToFieldBean>().select("field"));
        HashSet<String> set = new HashSet<>();
        fieldBeanList.forEach(fieldBean -> {
            if (set.add(fieldBean.getField())) {
                chatRecordService.deleteIndex(fieldBean.getField().toLowerCase());
                chatRecordService.createIndex(fieldBean.getField().toLowerCase(), ChatRecordBean.buildMapping());
                List<RoomBean> roomBeanList = roomMapper.selectList(new QueryWrapper<RoomBean>().select("room_name").eq("field", fieldBean.getField()));
                roomBeanList.forEach(roomBean -> {
                    List<GroupMessageBean> groupMessageBeanList = groupMessageMapper.selectList(new QueryWrapper<GroupMessageBean>().eq("field", fieldBean.getField()).eq("room_name", roomBean.getRoomName()));
                      BulkRequest bulkRequest = new BulkRequest(fieldBean.getField());
                    AtomicBoolean hasRequest = new AtomicBoolean(false);
                    groupMessageBeanList.forEach(groupMessageBean -> {
                        if (groupMessageBean.getMsgType() == 0) {
                            hasRequest.set(true);
                            ChatRecordBean chatRecordBean = new ChatRecordBean();
                            chatRecordBean.setField(fieldBean.getField());
                            chatRecordBean.setSenderId(groupMessageBean.getSenderId());
                            chatRecordBean.setRoomName(roomBean.getRoomName());
                            chatRecordBean.setSenderName(groupMessageBean.getSenderName());
                            chatRecordBean.setSendTime(groupMessageBean.getSendTime().toString());
                            chatRecordBean.setContent(groupMessageBean.getContent());
                            IndexRequest indexRequest = new IndexRequest(fieldBean.getField()).source(new Gson().toJson(chatRecordBean), XContentType.JSON);
                            bulkRequest.add(indexRequest);
                        }
                    });

                    try {
                        if(hasRequest.get()){
                            esClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
            }
        });
    }
}
